package com.lkh.sink;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import redis.clients.jedis.Jedis;

public class WordCountRedisSink extends RichSinkFunction<Tuple3<String, String, Integer>> {

    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        //打开redis连接

        jedis = new Jedis("master");

    }

    @Override
    public void invoke(Tuple3<String, String, Integer> value, Context context) throws Exception {

        String key = "word_cloud:" + value.f0;

        //将数据保存到redsi
        jedis.hset(key, value.f1, value.f2.toString());
    }

    @Override
    public void close() throws Exception {

        if (jedis!=null){
            jedis.close();
        }
    }
}
